package com.atguigu.edu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;


public class DwdTrafficUserJumpDetail {
    public static void main(String[] args) throws Exception {
        //TODO 1 获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        //设置并行度
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2 设置检查点和状态后端

/*    env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
    env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
    env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
    env.setStateBackend(new HashMapStateBackend());
    env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
    System.setProperty("HADOOP_USER_NAME", "atguigu");*/


        //读取kafka主题数据
        String groupId = "dwd_traffic_user_jump_detail";
        String topicName = "dwd_traffic_page_log";
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));

        //TODO 4 转换数据结构为JsonObject
        SingleOutputStreamOperator<JSONObject> map = streamSource.map(JSON::parseObject);

        // TODO 5 添加水位线watermark
        SingleOutputStreamOperator<JSONObject> beanWithWatermarkStream = map.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                }));

        //TODO 6 按照mid分组
        KeyedStream<JSONObject, String> keyedStream = beanWithWatermarkStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getJSONObject("common").getString("mid");
            }
        });

        //TODO 7 定义一个匹配规则pattern
        Pattern<JSONObject, JSONObject> objectPattern =
                Pattern.<JSONObject>begin("begin").where(new IterativeCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value, Context<JSONObject> ctx) throws Exception {
                String lastpageId = value.getJSONObject("page").getString("last_page_id");
                return lastpageId == null;
            }
        }).next("next")
                .where(new IterativeCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value, Context<JSONObject> ctx) throws Exception {
                        String lastpageId = value.getJSONObject("page").getString("last_page_id");
                        return lastpageId == null;
                    }
                }).within(Time.seconds(10L));

        //TODO 8 使用规则对数据流进行匹配
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, objectPattern);

        //TODO 9 拆分出超时流
        OutputTag<JSONObject> timeoutTag = new OutputTag<JSONObject>("timeout") {
        };
        SingleOutputStreamOperator<JSONObject> selectStream =
                patternStream.flatSelect(timeoutTag, new PatternFlatTimeoutFunction<JSONObject, JSONObject>() {
            @Override
            public void timeout(Map<String, List<JSONObject>> pattern, long timeoutTimestamp,
                                Collector<JSONObject> out) throws Exception {
                List<JSONObject> begin = pattern.get("begin");
                JSONObject jsonObject = begin.get(0);
                out.collect(jsonObject);
            }
        }, new PatternFlatSelectFunction<JSONObject, JSONObject>() {
            @Override
            public void flatSelect(Map<String, List<JSONObject>> pattern, Collector<JSONObject> out) throws Exception {
                List<JSONObject> begin = pattern.get("begin");
                JSONObject jsonObject = begin.get(0);
                out.collect(jsonObject);
            }
        });
        DataStream<JSONObject> timeoutStream = selectStream.getSideOutput(timeoutTag);

        // TODO 10 合并两个流
        DataStream<JSONObject> jumpStream = selectStream.union(timeoutStream);


        // TODO 11 写出到新的kafka主题
        String targetTopic = "dwd_traffic_user_jump_detail";
        jumpStream.map(JSONAware::toJSONString)
                .addSink(KafkaUtil.getKafkaProducer(targetTopic));

        jumpStream.print("jump>>>");

        // TODO  执行任务
        env.execute();
    }
}
